The present invention relates to certain challenges encountered when replicating data from one computer system to another. Using data replication, the database of a target computer system can be kept in synchronism with the information stored in the database of a source computer system. Alternatively, a target application can be fed database changes being made to a source database so that the target system can provide further processing services based on these events.
Database changes are typically replicated in groups called “transactions.” A transaction is a related group of database updates that either must be made at all systems, or else none of them are made. A problem occurs if the source system or if the network connecting the two systems fails during replication. The target system does not know whether or not it has received all updates for each transaction in progress and therefore does not know whether to apply the updates or to reject them. This condition is known as a hung transaction.
Data replication may be accomplished asynchronously. Changes are sent to the target database after they have been applied to the source database. The target database will lag the source database by a short delay, which is called replication latency.
Data replication also may be accomplished synchronously. In this case, the source application cannot complete a transaction to its database until it is assured that the changes associated with that transaction have been received by the target system. The source application is slowed by this delay, which is called application latency.
Furthermore, whether using asynchronous replication or synchronous replication, there may be a significant delay from when the source system issues a commit directive and the target system receives the commit directive. This slows down the target system because it must hold locks on data objects that the transaction has updated, thus slowing down other applications that need to update those data objects. Such a delay is called commit latency.
Computer Applications
Data-processing applications form the basis for much of our daily activities, from business to entertainment. Most applications are implemented as one or more programs running in a computer. In many cases, an application depends upon a database of information that the application maintains to record its current state. Typically, the information in the database is fundamental to the operation of the application, to the decisions it makes, and to its delivery of services to the end users.
The application's end users may include people, other applications, devices, and other systems. In this specification, the term “end user” means any entity that can influence an application and/or can use the services that it provides.
The fundamental components of an application are shown in FIG. 1. The application comprises a database and a program that is running in a computer. The database may be stored in persistent storage such as a disk for durability, it may be stored in high-speed memory for performance, or it may use a combination of these storage techniques. The database may be resident in the same computer as the application program, it may be resident in another computer, it may be implemented as an independent system, or it may be distributed among many systems.
A database generally includes several files or tables, though it may be just a random collection of unorganized data. Each file or table typically represents an entity set such as “employees” or “credit cards.” A file comprises records, each describing a member of an entity set such as an employee. A table comprises rows that describe members of an entity set. In this specification, “files” are equivalent to “tables;” and “records” are equivalent to “rows.”
With reference to FIG. 1, the application receives inputs from certain end users (1). It processes these inputs and may make certain modifications to its database as a result (2).
Database modifications are made via DML and DDL commands. DML (Data Manipulation Language) commands modify the contents of the database. Examples of DML commands are insert a row, update a row (modify its contents), and delete a row. DDL (Data Definition Language) commands typically modify the structure of the database. Examples of DDL commands include insert or delete a table and insert or delete a column in an existing table.
The application can read the contents of rows in its database (3). As part of its processing, it may read certain information from its database to make decisions. Based on the inputs it receives from its end users and the data in its database, the application delivers certain services to its end users (4). A service may be delivered as the result of a specific input from an end user, such as providing an account balance in response to an online banking query. Alternatively, a service may be delivered spontaneously by the application program, such as on a timed basis or when certain conditions occur. For instance, an alarm may be generated to operations staff if the load being carried by an electric-power transmission line exceeds a specified threshold.
The end users providing the input to the application may or may not be the same end users as those that receive its services.
Transactions
In many applications, changes to the database (for instance, inserts, updates, deletes, DDL changes to the database structure) are organized as transactions. A transaction is a delimited set of changes that must all be made to a database or sent to an application. Otherwise, none are made. For instance, a transaction in a banking application may transfer funds from one account to another. It applies a debit to one account (a reduction in its value) and an equal credit to another account (an increase in its value). Either both of these updates must occur or neither must occur in order to keep the customer's accounts balanced.
Transactions exhibit ACID properties—Atomicity, Consistency, Isolation, and Durability:                i. Atomicity means that either all operations contained within the transaction are executed against the database or none are.        ii. Consistency means that at any time, the view of the database represents an accurate view of the application data.        iii. Isolation means that a transaction is unaffected by other transactions that are executing simultaneously.        iv. Durability means that the resulting modification to the database by a transaction will survive any subsequent system failures.        
The changes comprising a transaction are delimited by a pair of directives. The beginning of a transaction is identified by a Begin Transaction directive (in some systems, the Begin Transaction directive is implied by the first change of a new transaction). The conclusion of a transaction is either a Commit Transaction directive or an Abort Transaction directive. A Commit Transaction directive causes all of the changes within the transaction to be applied. An Abort Transaction directive causes the changes within the transaction to be discarded. Though the terms Begin Transaction, Commit Transaction, and Abort Transaction are used in this specification, they are often known by different terms in different systems. However, the actions they denote are substantially the same in all systems.
A typical transaction appears as follows:                Begin Transaction        Insert Row A        Read Row B        Update Row B        Delete Row C        Commit Transaction        
The property of atomicity is guaranteed by ensuring that either all updates within the transaction are applied or none are.
The property of consistency is typically guaranteed by locking all data objects that are to be changed so that their value is hidden from other applications until they have been committed as an entire transaction.
The property of isolation is typically also guaranteed by locking all data objects that are to be changed so that no other transaction can modify their values until the current transaction commits.
The property of durability is typically guaranteed by writing the changes to a persistent database such as disk so that the changes survive any ensuing system failures.
In some implementations, the ACID properties are relaxed. For instance, consistency and isolation may be sacrificed to improve performance. An example of this is the BASE transaction protocol (see “Distributed Transactions,” Paul Krzyzanowski, Rutgers University; November 2012).
Data Replication
Many applications share data between two systems. This can be accomplished using data replication. A data replication engine accesses data changes generated by a source system and transfers them to a target system. The transfer may be made to a target database, in which case the replicated changes are applied to the target database. Alternatively, the transfer may be made to a target application, in which case the target application applies further processing to the replicated changes. Changes sent from the source system to a target database or application are also known as events.
A source application may send changes directly to the data replication engine, as shown in FIG. 2. In this configuration, each change that the source application generates is passed directly to the replication engine (1). The replication engine transfers the change to a database (2) or to another application (3) on the target system.
Alternatively, in many applications, one or more source applications make changes to a local source database. The replication engine follows these changes and replicates them to the target system database or applications. In order for changes to be passed to the replication engine, the changes are typically made available to the replication engine via a change queue. The change queue contains all of the database changes made by the various source applications. Though the changes from one application will be mixed with the changes from other applications, the order of changes made by one application is maintained in the change queue. The replication engine follows the insertion of changes made into the change queue and replicates them to the target system.
As illustrated in FIG. 3, there are several methods to create a change queue (1). The change queue may be created directly by the source application (2). As it makes changes to its source database, the source application also enters the changes into the change queue.
The change queue can be created from triggers in the source database (3). Whenever a database item is changed, a trigger in the database invokes a program (a stored procedure) that places the change in the change queue.
A common technique is to use the change log created by a transaction manager as the change queue. In most systems that support transactions, a transaction manager (4) handles the details of transaction processing, including data-object locks and begin/commit/abort directives. As part of its recovery mechanism, the transaction manager maintains a log of all changes (1) made to the database. This log can be used to reconstruct transactions that may have been lost due to a system failure. It can also be used to feed changes to the data replication engine.
The replication engine comprises an Extractor (5) and an Applier (6). It is the job of the Extractor to follow the change queue and to read the changes as they are entered into the change queue. The Extractor sends each change (or blocks of changes to improve communication-line utilization) over a communication channel (7) to the Applier, which applies the changes to the target database and/or passes them on to one or more target applications.
For increased performance, the various components of the replication engine can be multithreaded. There may by multiple Extractors, multiple communication channels, and multiple Appliers all sharing the replication load.
To facilitate heterogeneity so that replicated data can be applied to different types of databases and applications, the replication engine typically provides powerful format conversion capabilities so that the data format of source-system changes can be automatically converted to the data format required by the target system.
In many applications, replication is performed on transactions. Although individual changes are sent from the source system to the target system, the target system will hold them until it receives a transaction-completion directive from the source system. If this directive is a commit transaction, the target system will process the changes. If the directive is an abort transaction, the target system will reject the changes.
Asynchronous Replication
There are two fundamental types of data replication—asynchronous replication and synchronous replication. The replication methodology described with respect to FIG. 3 provides asynchronous replication. Changes are replicated after the fact from a change queue (1). The change queue is populated as the source application program makes changes to its local database. The change queue can be populated by the source application program (2), the source database (3), or by the transaction manager (4). After the change has added to the change queue, the replication engine reads the changes from the change queue (5), sends them over the communication channel (7) and applies them to the target database or forwards them to the target application program.
If transactions are replicated, they are typically managed at both the source system and target system by a transaction manager. The source system will begin a transaction, and the target system will begin a separate transaction when it receives the begin-transaction directive or the first change for a new transaction.
If replication is to a target database, the changes will be safe-stored by the target database or will be applied to the target database under the control of its transaction manager. When the transaction is completed on the source system, the transaction manager will insert a commit/abort token into the change queue. This token will be replicated to the target system, where it will cause the target system to either commit the transaction to the target database or to abort it, as appropriate.
If replication is to a target application, the application will safe-store the changes until it receives a transaction-completion directive via the replicated commit/abort token. If it is told to commit the transaction, the application will process the changes. If it is told to abort the transaction, it will ignore the changes.
The source application is unaware that replication is taking place and is unaffected by it. However, there is a delay, known as replication latency, from the time that the source system changes its database to the time that the change is received and is processed by the target system. Typical values for replication latency are tens of milliseconds to minutes, depending upon the implementation of the replication engine. If the source system should fail, any changes that are in the replication pipeline will be lost.
The Shadowbase asynchronous data replication engine from Gravic, Inc., is an embodiment of the asynchronous replication engine depicted in FIG. 3.
Synchronous Replication
With synchronous replication, no change can be completed at the source database until the target system has acknowledged that it has safe-stored the change or has applied it to its database. Thus, no data will be lost if the source system fails. The target system is guaranteed to have received all of the changes generated by the source system prior to its failure.
Network Transactions
A synchronous replication method known as Network Transactions is shown in FIG. 4. This method does not use a separate data replication engine. Rather, the replication logic is built into the source application. Each operation made by the source application to its local database is also made to the target database or is sent to the target application by the source application. The source application must wait for the operation to complete at both the source and target systems before it can declare the operation complete. If the operation successfully completes on only one of the systems, it is treated as a failed operation by the source application.
With respect to FIG. 4, the source system is connected to the target system via a communication channel (1). In order to execute a synchronously replicated transaction, the source system must first request that the transaction manager (2) begin a transaction (3a). The transaction manager knows that the target system is a party to the transaction, and it sends a begin-transaction directive (3b) to the target system (3c), thus allowing the target system to join the transaction.
As the source system makes a change to its local database (4a), it also sends that change (4b) over the communication channel to the target system (4c). The source system must wait for the target system to respond that it has safe-stored the change or has applied the change before it can continue to its next operation.
When all changes have been applied to both systems, the source application requests the transaction manager to commit or abort the transaction (5a). The transaction manager sends a commit-transaction or an abort-transaction directive (5b) to the target system (5c) so that the target system can vote on the transaction. If the source system asked for the transaction to be committed, and the target system votes “yes,” the transaction is committed on both systems. If the source system aborted the transaction, or if the target system votes “no,” the transaction is aborted on both systems.
Thus, the transaction is guaranteed to either be totally applied to both systems or be applied to neither system. This is the transaction attribute of atomicity as described earlier.
As a result of having to wait for each operation—the begin directive, each change, and the commit/abort directive—to applied on the target system, the source application slows down. The increased delay imposed upon the source application as the result of synchronous replication is known as application latency. Application latency increases as the distance between the source and target system increases because of the time that it takes for signals to move between the two systems. Consequently, synchronous replication with Network Transactions is typically limited to tens of miles, suitable for a campus or metropolitan area, Asynchronous replication, on the other hand, can be used over thousands of miles.
Coordinated Commits
Another method for synchronous replication is Coordinated Commits. Coordinate Commits uses a combination of the asynchronous and synchronous replication methods described above in the sections entitled “Asynchronous Replication” and “Synchronous Replication” to eliminate the challenges of data loss, to substantially reduce application latency, and to eliminate data collisions.
Coordinated Commits utilizes an asynchronous replication engine that joins a transaction initiated by the source application, as shown in FIG. 5. When the source application informs the transaction manager that it is starting a transaction (1a), it also informs the data replication engine (1b). The data replication engine registers with the transaction manager to tell the transaction manager that it is joining the transaction (2). In this way, it becomes a voting member of the transaction and can join in the determination as to whether the transaction should be committed or aborted.
At this point, the transaction manager can communicate with the target system by sending it events via the change queue over the replication channel. The transaction manager also can communicate directly with the replication engine and the target system without having to wait for events to propagate via the change queue. It does so by using “out-of-band” signals sent to the replication engine.
The begin directive and all database changes made by the source application to its database (3) are asynchronously replicated to the target system via the change queue. As the source database is modified, the changes are captured in the change queue (4). The replication engine fetches the changes from the change queue (5) and replicates them to the target system (6), as described earlier in the section entitled “Asynchronous Replication”.
The completion of the transaction is shown in FIG. 6. When the source application is ready to commit the transaction, it issues a commit directive to the transaction manager (1). The transaction manager will ask all of the parties that have joined the transaction to vote on whether or not they are ready to commit (RTC?) the transaction. In the case of FIG. 6, the parties are the source database system (2a) and the data replication engine (2b), which receives an RTC? signal from the transaction manager. If the transaction has been safe-stored on the source database, the source database system will respond with a “yes” (3a). If all changes have been replicated and safe-stored on the target system, the replication engine will vote “yes” to the commit request (3b).
If all voting parties respond positively, the transaction manager will tell the source database system to commit the transaction (4a) and optionally tell the replication engine to commit the target transaction via a commit signal (4b). The transaction manager will write a commit token into the change queue (4c). At this point, the transaction manager will inform the application that the transaction has committed (5). If instead, any party to the transaction responds negatively, the transaction manager will tell all parties to the transaction to abort the transaction.
If the transaction manager has signaled the replication engine to commit or abort the transaction (4b), the target system will commit or abort the transaction and will ignore the commit token when it is later received over the replication channel (4c). If the replication engine is not informed of the commit or abort by a transaction manager signal, the target system will commit or abort the transaction when it receives the commit token over the replication channel.
Note that the source system does not have to wait until the transaction is actually committed at the target system. The source system knows that the transaction will complete as soon as the target system receives the commit signal from the transaction manager or receives the commit token over the replication channel.
Thus, with the Coordinated Commits method, changes are replicated asynchronously; but the commit or abort completion of the transaction is coordinated between the source and target systems. The transaction will either be committed on both systems, or it will be committed on neither.
The time that the source system must wait for the target system to vote and then to receive a commit directive adds to the transaction processing time of both the source and target systems. This additional time is called application latency and is comprised in large part by the additional commit time, which is called commit latency. The target-system commit latency is significantly reduced if it gets its commit directive directly from the transaction manager as a signal rather than having to wait for the commit token to be replicated. Commit latency can be reduced by other enhancements as well. For instance, if the target system responds to replicated updates with an acknowledgement signal to the source system, the source system will know in advance whether the target system is ready to commit without having to go through the target system's voting process. If the target system has acknowledged the receipt of the last update, the source system knows that the target system has received and safe-stored or applied all of the transaction updates. This is an implied “yes” vote, and the replication-engine voting sequence can be bypassed.
Since the source system must only wait for the target system to acknowledge that all of the data for the transaction has been safe-stored, the application latency that Coordinated Commits imposes on the source application is minimal. Thus, the source and target nodes can be separated by thousands of miles rather than by tens of miles.
With Coordinated Commits, there is no data loss if the source system or replication network fails, as transactions are either committed or aborted on both systems. Application latency is greatly reduced because the source application only has to wait for the commit response from the target system. It does not have to wait on a round-trip network communication on every database change before proceeding.
Bidirectional Replication
In some applications, all processing nodes are active. All the nodes are processing transactions. A transaction can be sent to any node in the application network and can be processed in the same way that it would be if it were sent to another node. This mode of operation is known as active/active processing.
In order for each node to properly process a transaction, all nodes must have the same view of the application database. This is achieved by replicating a change made by any one node to every other node in the application network. Since changes are replicated in either direction between two nodes, this is known as bidirectional replication.
Bidirectional replication imposes a new challenge in addition to asynchronous replication data loss or synchronous replication application latency. This is the problem of data collisions.
In an asynchronous bidirectional replication system, it is possible for two separate nodes to update the same data object within the replication latency interval. Each change will be replicated to the other node and will overwrite the change originally made at that node. Now, both nodes have a different value for the data object, and both are wrong. There are methods for resolving data collisions automatically by establishing rules that assign a priority to the changes (such as the earliest change is accepted). If automatic resolution is not possible, the collision must be resolved manually.
A similar challenge occurs with synchronous bidirectional replication except that data collisions potentially become deadlocks, during which time neither system can proceed. Deadlocks can occur if both systems acquire locks in a different order. For instance, assume that as part of a pair of transactions, System A wants to update Rows 1 and 2. System B, on the other hand, wants to update Rows 2 and 1, in those orders. Each system must acquire a lock on its row before it can update it (the transaction attributes of consistency and isolation, as described earlier). System A will acquire a lock on Row 1, and System B will acquire a lock on Row 2. Now System A cannot update Row 2, and System B cannot update Row 1. The two systems are deadlocked. A means must be provided to detect and resolve deadlocks, perhaps by having one or both systems release their locks and try again later.